package com.sui.bigdata.sml.web.misc.k8s;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.sui.bigdata.sml.web.configuration.KubeConfig;
import com.sui.bigdata.sml.web.util.Constants;
import io.kubernetes.client.ApiException;
import io.kubernetes.client.informer.ResourceEventHandler;
import io.kubernetes.client.informer.SharedIndexInformer;
import io.kubernetes.client.informer.SharedInformerFactory;
import io.kubernetes.client.models.*;
import io.kubernetes.client.util.CallGeneratorParams;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author songhaicheng
 * @date 2020/3/18 15:14
 * @description
 * @reviewer
 */
@Slf4j
public class KubeClient {

    private static final SharedInformerFactory INFORMER_FACTORY;

    private static final List<String> PHASES = Arrays.asList("Running", "CrashLoopBackOff");
    private static final ConcurrentHashMap<String, Evolve> DPLY_REPLICATING = new ConcurrentHashMap<>();

    static {
        // 注册 Pod 组件监听方法
        INFORMER_FACTORY = new SharedInformerFactory();
        SharedIndexInformer<V1Pod> informer = INFORMER_FACTORY.sharedIndexInformerFor(
                (CallGeneratorParams params) -> KubeConfig.serviceAPI.listNamespacedPodCall(KubeConfig.NAMESPACE, null, null, null, null, null, params.resourceVersion, params.timeoutSeconds, params.watch, null, null),
                V1Pod.class, V1PodList.class);
        informer.addEventHandler(new ResourceEventHandler<V1Pod>() {
            @Override
            public void onAdd(V1Pod obj) {
                log.info("onAdd "+obj.getMetadata().getName()+",status:"+obj.getStatus().getPhase()+",isready："+obj.getStatus().getContainerStatuses().stream().allMatch(V1ContainerStatus::isReady));
                updatePodStatus(obj);
            }
            @Override
            public void onUpdate(V1Pod oldObj, V1Pod newObj) {
                // 目标 Pod 返回成功或错误状态
                log.info("onUpdate "+newObj.getMetadata().getName()+",status:"+newObj.getStatus().getPhase()+",isready："+newObj.getStatus().getContainerStatuses().stream().allMatch(V1ContainerStatus::isReady));
                updatePodStatus(newObj);
            }
            @Override
            public void onDelete(V1Pod obj, boolean deletedFinalStateUnknown) {
                log.info("onDelete "+obj.getMetadata().getName()+",status:"+obj.getStatus().getPhase()+",isready："+obj.getStatus().getContainerStatuses().stream().allMatch(V1ContainerStatus::isReady));
            }
        });
        INFORMER_FACTORY.startAllRegisteredInformers();
    }

    private static void updatePodStatus(V1Pod obj) {
        for (Map.Entry<String, Evolve> entry : DPLY_REPLICATING.entrySet()) {
            if (obj.getMetadata().getName().startsWith(entry.getKey() + "-") &&
                    PHASES.contains(obj.getStatus().getPhase())) {
                if (PHASES.indexOf(obj.getStatus().getPhase()) == 0) {
                    if (obj.getStatus().getContainerStatuses().stream().allMatch(V1ContainerStatus::isReady)) {
                        entry.getValue().currentPlus();
                    }
                } else {
                    // 一个 replica 出错，则强行认为此次启动失败
                    entry.getValue().setSuccess(false);
                }
            }
        }
    }

    public void createDeployment(V1Deployment deployment) throws ApiException {
        if (!DPLY_REPLICATING.containsKey(deployment.getMetadata().getName())) {
            DPLY_REPLICATING.putIfAbsent(
                    deployment.getMetadata().getName(),
                    new Evolve(CustomConfiguration.getInt(Constants.KUBE_DEPLOYMENT_REPLICAS))
            );
            KubeConfig.deploymentAPI.createNamespacedDeployment(KubeConfig.NAMESPACE, deployment, null, null, null);
        }
    }

    public V1Deployment replaceDeployment(String servingToken, V1Deployment deployment) throws ApiException {
        return KubeConfig.deploymentAPI.replaceNamespacedDeployment(KubeConfig.NAME_PREFIX + servingToken, KubeConfig.NAMESPACE, deployment, null, null, null);
    }

    public void deleteDeployment(String name) throws ApiException {
        KubeConfig.deploymentAPI.deleteNamespacedDeployment(KubeConfig.NAME_PREFIX + name, KubeConfig.NAMESPACE, null, null, null, null, null, null);
    }

    public void createService(V1Service service) throws ApiException {
        KubeConfig.serviceAPI.createNamespacedService(KubeConfig.NAMESPACE, service, null, null, null);
    }

    public void deleteService(String name) throws ApiException {
        KubeConfig.serviceAPI.deleteNamespacedService(KubeConfig.NAME_PREFIX + name, KubeConfig.NAMESPACE, null, null, null, null, null, null);
    }

    public void replaceService(String servingToken, V1Service service) throws ApiException {
        KubeConfig.serviceAPI.replaceNamespacedService(KubeConfig.NAME_PREFIX + servingToken, KubeConfig.NAMESPACE, service, null, null, null);
    }

    public V1ServiceList listService(String servingToken) throws ApiException {
        return KubeConfig.serviceAPI.listNamespacedService(KubeConfig.NAMESPACE, null, null, null, KubeConfig.NAME_PREFIX + servingToken, null, null, null, null);
    }

    public void createIngress(ExtensionsV1beta1Ingress ingress) throws ApiException {
        KubeConfig.ingressAPI.createNamespacedIngress(KubeConfig.NAMESPACE, ingress, null, null, null);
    }

    public void deleteIngress(String name) throws ApiException {
        KubeConfig.ingressAPI.deleteNamespacedIngress(KubeConfig.INGRESS_NAME_PREFIX + name, KubeConfig.NAMESPACE, null, null, null, null, null, null);
    }

    public void replaceIngress(String servingToken, ExtensionsV1beta1Ingress ingress) throws ApiException {
        KubeConfig.ingressAPI.replaceNamespacedIngress(KubeConfig.NAME_PREFIX + servingToken, KubeConfig.NAMESPACE, ingress, null, null, null);
    }

    public ExtensionsV1beta1IngressList listIngress(String servingToken) throws ApiException {
        return KubeConfig.ingressAPI.listNamespacedIngress(KubeConfig.NAMESPACE, null, null, null, KubeConfig.NAME_PREFIX + servingToken, null, null, null, null);
    }

    public List<String> listInstance(String servingToken) throws ApiException {
        return KubeConfig.serviceAPI
                .listNamespacedPod(KubeConfig.NAMESPACE, null, null, null, null, null, null, null, false)
                .getItems().stream().map(item -> item.getMetadata().getName())
                .filter(name -> name.startsWith(KubeConfig.NAME_PREFIX + servingToken))
                .collect(Collectors.toList());
    }

    public boolean testEntity(String servingToken) {
        String name = KubeConfig.NAME_PREFIX + servingToken;
        Evolve evolve = DPLY_REPLICATING.get(name);
        long start = System.currentTimeMillis();
        // 设置监听超时时间。此处仅为备用方案，防止线程卡死，尽量交由 K8S 通知实体是否创建成功
        long timeout = CustomConfiguration.getInt(Constants.KUBE_SML_READINESS_SECONDS_TIMEOUT, 60) * 1000;
        // K8S 监听间隔为 T * 0.2，这里设置比间隔时间稍少些，这样可以不错过来自 K8S 的通知
        long interval = timeout / 6;
        // 堵塞住直到 K8S 操作完毕
        while (!evolve.isDone()) {
            try {
                Thread.sleep(interval);
                if (System.currentTimeMillis() - start >= timeout) {
                    evolve.setSuccess(false);
                }
            } catch (InterruptedException e) {
                log.error("尝试监听 K8S 状态失败：", e);
                evolve.setSuccess(false);
            }
        }
        boolean success = evolve.isSuccess();
        DPLY_REPLICATING.remove(name);
        return success;
    }
}
